package com.lihh;

import org.apache.zookeeper.*;
import org.apache.zookeeper.data.Stat;

import java.io.IOException;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.CountDownLatch;

/**
 * 实现 分布式锁
 *
 * @author lihh
 */
public class DistributedLock {
  // 连接 zookeeper 客户端
  private ZooKeeper client;
  // zookeeper 服务集群地址
  private final static String connectString = "192.168.56.17:2181,192.168.56.18:2181,192.168.56.19:2181";
  // 超时时间
  private int sessionTimeOut = 30000;
  // 等待zk 连接成功
  private CountDownLatch connectLatch = new CountDownLatch(1);
  // 等节点变化
  private CountDownLatch waitLatch = new CountDownLatch(1);
  // 当前节点
  private String currentNode;
  // 前一个节点路径
  private String waitPath;
  
  public DistributedLock() throws IOException, InterruptedException, KeeperException {
    client = new ZooKeeper(connectString, sessionTimeOut, new Watcher() {
      @Override
      public void process(WatchedEvent watchedEvent) {
        // 表示 连接成功的状态
        if (watchedEvent.getState() == Event.KeeperState.SyncConnected) {
          connectLatch.countDown();
        }
        
        // 表示节点被删除 && 删除的前一个节点
        if (watchedEvent.getType() == Event.EventType.NodeDeleted && watchedEvent.getPath().equals(waitPath)) {
          waitLatch.countDown();
        }
      }
    });
    
    // 等待连接成功
    connectLatch.await();
    
    // 2. 判断节点是否存在
    Stat stat = client.exists("/locks", false);
    if (stat == null) {
      client.create("/locks", "locks".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
    }
  }
  
  /**
   * 进行加锁
   *
   * @author lihh
   */
  public void lock() throws Exception {
    // 创建临时节点
    currentNode = client.create("/locks/seq-", null, ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);
    
    // 拿到所有的子节点
    List<String> children = client.getChildren("/locks", false);
    // 如果只有一个节点的话，直接拿到锁
    if (children.size() == 1) {
      return;
    }
    
    // 排序（序号小的 先拿到锁）
    Collections.sort(children);
    // 获取节点名称
    String nodeName = currentNode.substring("/locks/".length());
    
    // 拿到 名称的位置
    int index = children.indexOf(nodeName);
    if (index == -1) {
      throw new Exception("数据异常了");
    } else if (index == 0) {
    } else {
      //需要监听前一个节点变化
      waitPath = "/locks/" + children.get(index - 1);
      client.getData(waitPath, true, null);
      
      //等待监听执行
      waitLatch.await();
    }
  }
  
  /***
   * 进行解锁
   *
   * @author lihh
   */
  public void unLock() throws InterruptedException, KeeperException {
    client.delete(currentNode, -1);
  }
}
